package com.shujia.flink.core;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

public class Demo03BatchMode {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        /*
         * 可以指定三种方式：
         * STREAMING：默认的运行方式，既可以作用在无界流上也可以作用在有界流上，结果是连续变化的
         * BATCH：只能作用在有界流上，以批处理方式处理数据，只能看到最终的结果
         * AUTOMATIC：自动识别，根据环境判断到底以STREAMING还是BATCH模式运行
         */
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 统计每个班级学生的平均年龄
        ArrayList<String> arr = new ArrayList<>();
        arr.add("文科一班,20");
        arr.add("文科一班,21");
        arr.add("文科一班,22");
        arr.add("文科二班,24");
        arr.add("文科二班,22");
        arr.add("文科二班,23");

        // 有界流：既可以流处理也可以批处理
        DataStreamSource<String> stuDS = env.fromCollection(arr);

        stuDS.map(s -> Tuple3.of(s.split(",")[0], Integer.parseInt(s.split(",")[1]), 1), Types.TUPLE(Types.STRING, Types.INT, Types.INT))
                .keyBy(t3 -> t3.f0, Types.STRING)
                .reduce((t1, t2) -> Tuple3.of(t1.f0, t1.f1 + t2.f1, t1.f2 + t2.f2))
                .map(t3 -> Tuple2.of(t3.f0, t3.f1 * 1.0 / t3.f2),Types.TUPLE(Types.STRING,Types.DOUBLE))
                .print();


        // Socket 是无界流：只能以STREAMING模式运行
//        env.socketTextStream("master",8888).print();

        env.execute();
    }
}
